package com.zhang.first.day04;

import com.zhang.first.utils.ClickEvent;
import com.zhang.first.utils.ClickEventSource;
import com.zhang.first.utils.UrlViewCountPerWindow;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * 每个url在每个5秒钟的滚动窗口中被浏览的次数
 * @title: 使用KeyedProcessFunction 实现增量聚合函数和全窗口聚合函数结合使用
 * @author: zhang
 * @date: 2022/1/17 20:43
 */
public class Example6 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        env
                .addSource(new ClickEventSource())
                .keyBy(r -> r.url)
                .process(new KeyedProcessFunction<String, ClickEvent, UrlViewCountPerWindow>() {
                    //key：windowStartTime
                    //value：url在windowStartTime对应的窗口中的浏览次数，也就是累加器
                    private MapState<Long, Long> accCountMap;
                    // 滚动窗口长度
                    private long windowSize = 5000L;

                    @Override
                    public void open(Configuration parameters) throws Exception {
                        accCountMap = getRuntimeContext().getMapState(new MapStateDescriptor<Long, Long>(
                                "accCountMap",
                                Types.LONG,
                                Types.LONG
                        ));

                    }

                    @Override
                    public void processElement(ClickEvent value, KeyedProcessFunction<String, ClickEvent, UrlViewCountPerWindow>.Context ctx, Collector<UrlViewCountPerWindow> out) throws Exception {
                        Long currTs = ctx.timerService().currentProcessingTime();
                        Long windowStartTime = currTs - currTs % windowSize;
                        Long windowEndTime = windowStartTime + windowSize;

                        if (!accCountMap.contains(windowStartTime)) {
                            accCountMap.put(windowStartTime, 1L);
                        } else {
                            accCountMap.put(windowStartTime, accCountMap.get(windowStartTime) + 1L);
                        }
                        ctx.timerService().registerProcessingTimeTimer(windowEndTime - 1L);
                    }

                    @Override
                    public void onTimer(long timestamp, OnTimerContext ctx, Collector<UrlViewCountPerWindow> out) throws Exception {
                        Long windowEndTime = timestamp + 1L;
                        Long windowStartTime = windowEndTime - windowSize;
                        String url = ctx.getCurrentKey();
                        Long count = accCountMap.get(windowStartTime);
                        out.collect(new UrlViewCountPerWindow(
                                url,
                                count,
                                windowStartTime,
                                windowEndTime
                        ));
                        // 销毁窗口
                        // .remove只清楚windowStartTime对应的键值对
                        // .clear将整个mapState清空
                        accCountMap.remove(windowStartTime);
                    }
                })
                .print();

        env.execute();
    }
}
